package org.yuanzheng.source

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.common.serialization.StringDeserializer

/**
 * @author yuanzheng
 * @date 2020/6/11-8:40
 */
object KafkaSourceString {
  def main(args: Array[String]): Unit = {
    // 1.初始化流计算环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 2.修改并行度
    streamEnv.setParallelism(1)
    // 3.导入隐式转换
    import org.apache.flink.streaming.api.scala._
    // 4.连接Kafka数据源
    val props = new Properties()
    props.setProperty("bootstrap.servers", "192.168.1.8:9092")
    props.setProperty("group.id", "test-consumer-group")
    props.setProperty("key.deserializer", classOf[StringDeserializer].getName)
    props.setProperty("value.deserializer", classOf[StringDeserializer].getName)
    props.setProperty("auto.offset.reset", "latest")
    // 5.配置
    val result = streamEnv.addSource(new FlinkKafkaConsumer[String]("yuan", new SimpleStringSchema(), props))
    // 6.
    result.print()
    //7.
    streamEnv.execute()


  }

}
